#include "coin2/exchange/deribit_base/api_order/fix/fix_trade_client2.h"

#include <chrono>
#include <filesystem>
#include <thread>
#include <utility>

#include "coin2/base/base64.h"
#include "coin2/exchange/base/api_util/api_signature.h"
#include "coin2/exchange/base/api_util/auth_key.h"
#include "coin2/exchange/deribit_futures/order_v2/parser2.h"
#include "coin2/exchange/deribit_futures/symbology/product.h"
#include "coin2/exchange/deribit_options/symbology/product.h"
#include "naviasync/ipc/asio_forwarder.h"
#include "navibase/read_config.h"

namespace coin2::exchange::deribit::api_order::fix {

namespace {

namespace resp_fix_hdr {

#define FIX_PARSER_FIX_HDR_TAG_TRAITS(TMPL) TMPL(tag::common::MsgSeqNum, int64_t)
DEF_RESP_PARSER(FixHdrParser, FIX_PARSER_FIX_HDR_TAG_TRAITS);

}  // namespace resp_fix_hdr

namespace resp_security_list {

#define FIX_PARSER_SECURITY_LIST_TAG_TRAITS(TMPL) \
  TMPL(tag::common::SecurityRequestResult)        \
  TMPL(tag::common::NoRelatedSym, int)
DEF_RESP_PARSER(SecurityListParser, FIX_PARSER_SECURITY_LIST_TAG_TRAITS);

}  // namespace resp_security_list

namespace rpt_grp_related_sym {

#define FIX_RPT_GRP_RELATED_SYMB_TAG_TRAITS(TMPL) \
  TMPL(tag::common::Symbol)                       \
  TMPL(tag::common::ContractMultiplier, int)

DEF_RESP_PARSER(RelatedSymParser, FIX_RPT_GRP_RELATED_SYMB_TAG_TRAITS);

}  // namespace rpt_grp_related_sym

struct AsioSequencerTraits {
  template <class Reactor>
  using ReactorBaseT = ::naviasync::ipc::AsioReactor<Reactor>;
};

auto BuildDefaultStreamClientContext(
    const FixTradeClient2::Context& ctx,
    const std::string& begin_string,
    const std::string& target_comp_id,
    int32_t plant_hb_interval_secs) {
  FixTradeClient2::StreamClient::Context stream_client_ctx;
  stream_client_ctx.raw_client_ctx.mutable_lws_client_common()->set_order_gateway_id(
      ctx.trade_client_ctx.og_id());
  stream_client_ctx.raw_client_ctx.mutable_lws_client_common()->set_connection_idx(
      ctx.trade_client_ctx.conn_idx());
  stream_client_ctx.sequencer_mixin_ctx.sender_comp_id = ctx.sender_comp_id;
  stream_client_ctx.sequencer_mixin_ctx.target_comp_id = target_comp_id;
  stream_client_ctx.sequencer_mixin_ctx.ver_str = begin_string;
  stream_client_ctx.sequencer_mixin_ctx.max_len_msg_type = 4;
  stream_client_ctx.sequencer_mixin_ctx.og_id = 1;
  stream_client_ctx.sequencer_mixin_ctx.conn_idx = 0;
  stream_client_ctx.sequencer_mixin_ctx.plant_hb_interval_secs = plant_hb_interval_secs;
  {
    static constexpr auto kHbMsgTmplSize = 128;
    stream_client_ctx.sequencer_mixin_ctx.hb_msg_tmpl = std::string(kHbMsgTmplSize, '\0');
    ::naviasraw::fix::Request heartbt(begin_string, "0", ctx.sender_comp_id, target_comp_id, "");
    heartbt.Bind(stream_client_ctx.sequencer_mixin_ctx.hb_msg_tmpl.data(), kHbMsgTmplSize);
    ::naviasraw::IdGenerator<> hb_tmpl_seq_no;
    heartbt.Commit(hb_tmpl_seq_no);
  }
  stream_client_ctx.own_receiver_ctx.set_busy_loop(false);
  return stream_client_ctx;
}

auto BuildDefaultSequencerContext(const std::string& main_reactor_name) {
  typename FixTradeClient2::StreamClient::OwnSequencerContextT<AsioSequencerTraits> sequencer_ctx;
  sequencer_ctx.set_topic_poll_mode(::naviasync::ipc::PollReactorContextProto::ROUND_ROBIN);
  sequencer_ctx.mutable_reactor()->set_name(main_reactor_name + ".sequencer");
  sequencer_ctx.mutable_reactor()->set_is_embedded(true);
  return sequencer_ctx;
}

}  // namespace

FixTradeClient2::FixTradeClient2(std::string_view auth_key_file, const MarketExchangeApi& mea)
    : mea_{mea},
      ctx_([auth_key_file] {
        Context ctx;
        CHECK(
            ReadJsonConfig(std::filesystem::path(kTradeClientConfig), &ctx.trade_client_ctx).ok());
        auto auth_key = AuthKey::FromFile(std::string(auth_key_file));
        ctx.sender_comp_id = auth_key.access_key();
        ctx.access_key = auth_key.access_key();
        ctx.secret_key = auth_key.secret_key();
        return ctx;
      }()),
      fix_msg_slicer_(
          mea_,
          kBeginString,
          [this](bool is_logon) {
            is_logon_ = is_logon;
            LOG(INFO) << mea_.String() << (is_logon ? " FIX logon." : " FIX logout.");
          },
          [this](int64_t new_seq_no) { next_recv_seq_no_ = new_seq_no; },
          [this](std::string_view msg_type, const base::order::BaseParser::RawMessage& app_msg) {
            if (this->ParseAppMessage(msg_type, app_msg.raw_buffer)) {
              processor_->ProcessRawMessage(app_msg);
            }
          }) {
  msg_builder_ =
      std::make_unique<FixMessageBuilder2>(mea_, kBeginString, ctx_.sender_comp_id, kTargetCompId);
  CHECK(msg_builder_ != nullptr);
}

void FixTradeClient2::Start() {
  if (is_inited_) return;
  is_inited_ = true;
  if (executor_ != nullptr) {
    timer_ = std::make_unique<boost::asio::steady_timer>(*executor_);
    timer_->expires_after(std::chrono::seconds(10));
    timer_->async_wait(std::bind(&FixTradeClient2::OnTimer, this, std::placeholders::_1));
  }

  CHECK_NOTNULL(main_reactor_);
  constexpr std::string_view kOgSndSuffix(".lws1.og_snd");
  constexpr std::string_view kOgRcvSuffix(".lws1.og_rcv");
  StreamClient::EmbeddedForwarder::Context ctx;
  forwarder_grp_ = std::make_unique<ForwarderGroup>();
  using TopicList = ::naviasync::ipc::TopicList;
  for (TopicList::TopicId topic_id = 0; topic_id < main_reactor_->topic_list()->num_topics();
       ++topic_id) {
    auto topic_name = main_reactor_->topic_list()->GetTopicName(topic_id);
    if (topic_name.size() > kOgSndSuffix.size() &&
        topic_name.substr(topic_name.size() - kOgSndSuffix.size(), kOgSndSuffix.size())
                .compare(kOgSndSuffix) == 0) {
      ctx.set_topic_name(topic_name);
      ctx.set_signal(naviasync::ipc::OG_SEND_RAW);
      forwarder_grp_->AddForwarder(ctx, main_reactor_->plant(), og_raw_ioc_);
    }
  }
  forwarder_grp_->SetNumSendForwarders();
  for (TopicList::TopicId topic_id = 0; topic_id < main_reactor_->topic_list()->num_topics();
       ++topic_id) {
    auto topic_name = main_reactor_->topic_list()->GetTopicName(topic_id);
    if (topic_name.size() > kOgRcvSuffix.size() &&
        topic_name.substr(topic_name.size() - kOgRcvSuffix.size(), kOgRcvSuffix.size())
                .compare(kOgRcvSuffix) == 0) {
      ctx.set_topic_name(topic_name);
      ctx.set_signal(naviasync::ipc::OG_RECV_RAW);
      forwarder_grp_->AddForwarder(ctx, main_reactor_->plant(), og_raw_ioc_);
    }
  }

  stream_client_ = std::make_unique<StreamClient>(
      BuildDefaultStreamClientContext(
          ctx_,
          kBeginString,
          kTargetCompId,
          main_reactor_->plant().heartbeat_interval_sec()),
      *main_reactor_,
      *forwarder_grp_,
      AsioSequencerTraits{},
      BuildDefaultSequencerContext(main_reactor_->name()),
      main_reactor_->plant(),
      og_raw_ioc_);
  main_reactor_->SetOgRawRespCallback(
      ctx_.trade_client_ctx.og_id(),
      ctx_.trade_client_ctx.conn_idx(),
      [this](const char* payload, size_t payload_len, int64_t ts) {
        if (__glibc_unlikely(payload_len == 0)) {
          LOG(WARNING) << "FIX line disconnected";
          is_logon_ = false;

          this->EstablishSession();
        }
        fix_msg_slicer_.ConsumeBuffer(payload, payload_len, ts);
      });

  std::thread([this] {
    main_reactor_->WaitUntilRegistered();
    stream_client_->res_mgr(*main_reactor_->res_mgr());
    stream_client_->EstablishConnFast(
        ctx_.trade_client_ctx.server(),
        ctx_.trade_client_ctx.service(),
        *forwarder_grp_,
        15);  // todo: setup retry intv
    this->EstablishSession();
  }).detach();
}

void FixTradeClient2::Stop() {
  CHECK(stream_client_->Disconnect().ok());
  is_logon_ = false;
}

void FixTradeClient2::OnTimer(const boost::system::error_code& ec) {
  int interval = 10;
  if (is_logon_) {
    char* msg_buf = nullptr;
    auto buf_size = stream_client_->GetSendMsgBuf(msg_buf);
    auto message = msg_builder_->GenerateSecurityListRequest(msg_buf, buf_size);
    if (message == nullptr) {
      LOG(ERROR) << "failed to generate logon";
      return;
    }
    stream_client_->Send(
        StreamClient::SendObj::Raw{message->GetBufStart(), message->GetBufSize()},
        *forwarder_grp_);
    interval = 900;
  }
  timer_->expires_after(std::chrono::seconds(interval));
  timer_->async_wait(std::bind(&FixTradeClient2::OnTimer, this, std::placeholders::_1));
}

bool FixTradeClient2::IsAvailable() const { return is_inited_ && is_logon_; }

void FixTradeClient2::SetExecutor(Executor* executor) { executor_ = executor; }

void FixTradeClient2::SetMainReactor(base::executor::CoinReactor& main_reactor) {
  main_reactor_ = &main_reactor;
}

void FixTradeClient2::SetProcessor(Processor* processor) { processor_ = processor; }

void FixTradeClient2::AddSubscriber(SubscriberCB cb) {}

void FixTradeClient2::PlaceOrder(const OrderExecution& ord_exec) {
  DCHECK(msg_builder_);
  auto contract_multiplier = GetContractValue(
      ord_exec.order_spec->product,
      ord_exec.order_spec->order_created_time,
      mea_,
      GetProductHolder());
  char* msg_buf = nullptr;
  auto buf_size = stream_client_->GetSendMsgBuf(msg_buf);
  auto message = msg_builder_->GeneratePlaceOrder(
      msg_buf,
      buf_size,
      ord_exec,
      GetProductHolder(),
      contract_multiplier);
  if (!message) {
    LOG(ERROR) << "failed to generate place order: " << ord_exec.proc_order_id;
    return;
  }
  stream_client_->Send(
      StreamClient::SendObj::Raw{message->GetBufStart(), message->GetBufSize()},
      *forwarder_grp_);
}

void FixTradeClient2::CancelOrder(const OrderExecution& ord_exec) {
  DCHECK(msg_builder_);
  char* msg_buf = nullptr;
  auto buf_size = stream_client_->GetSendMsgBuf(msg_buf);
  auto message = msg_builder_->GenerateCancelOrder(msg_buf, buf_size, ord_exec);
  if (!message) {
    LOG(ERROR) << "failed to generate cancel order: " << ord_exec.proc_order_id;
    return;
  }
  stream_client_->Send(
      StreamClient::SendObj::Raw{message->GetBufStart(), message->GetBufSize()},
      *forwarder_grp_);
}

auto FixTradeClient2::GetContractValue(
    const std::string& native_product,
    int64_t ts,
    const MarketExchangeApi& mea,
    GetProductHolderFunc get_product_holder) -> double {
  if (contract_value_map_.count(native_product) == 0) {
    if (mea.market == coin::proto::Futures) {
      using coin2::exchange::deribit_futures::symbology::DeribitFuturesProduct;
      auto product = DeribitFuturesProduct::FromStr(native_product, ts);
      auto product_holder = get_product_holder(product, ts);
      // daniel(20230803): for Deribit futures, the contract value always is set as 1
      //                   and the qty ticksize is set as contract size when retrieve
      //                   pi from exchange
      contract_value_map_[native_product] = product_holder->product_info().tick_qty().TickSize();
    } else if (mea.market == coin::proto::Options) {
      using coin2::exchange::deribit_options::symbology::DeribitOptionsProduct;
      auto product = DeribitOptionsProduct::FromStr(native_product, ts);
      auto product_holder = get_product_holder(product, ts);
      contract_value_map_[native_product] = product_holder->product_info().contract_value();
    } else {
      CHECK(false) << "Unknown market type " << mea.String();
    }
  }
  return contract_value_map_.at(native_product);
}

void FixTradeClient2::SetContractValue(const std::string& native_product, double val) {
  if (contract_value_map_[native_product] == val) {
    return;
  }
  SPDLOG_INFO(
      "reset contract multiplier for symbol {}: {} -> {}",
      native_product,
      contract_value_map_[native_product],
      val);
  contract_value_map_[native_product] = val;
}

bool FixTradeClient2::ParseAppMessage(std::string_view msg_type, std::string_view message) {
  constexpr auto kNeedFurtherParsing = true;

  using namespace ::naviasraw::fix;
  resp_fix_hdr::FixHdrParser fix_hdr_parser;
  auto success = fix_hdr_parser.Parse(message);
  DCHECK(success) << "Failed to parse " << message;
  next_recv_seq_no_ += fix_msg_slicer_.RetrieveNumSkippedMsg();
  if (__glibc_unlikely(next_recv_seq_no_ != fix_hdr_parser.GetVal(tag::common::MsgSeqNum()))) {
    LOG(ERROR) << "MsgSeqNum mismatch! Receiverd: "
               << fix_hdr_parser.GetVal(tag::common::MsgSeqNum())
               << " Expect: " << next_recv_seq_no_;
    CHECK(stream_client_->Disconnect().ok());
    is_logon_ = false;
    stream_client_->EstablishConnFast(
        ctx_.trade_client_ctx.server(),
        ctx_.trade_client_ctx.service(),
        *forwarder_grp_,
        15);  // todo: setup retry intv
    return false;
  }
  ++next_recv_seq_no_;

  if (msg_type == "y") {
    resp_security_list::SecurityListParser sec_list_parser;
    rpt_grp_related_sym::RelatedSymParser rpt_grp_parser;
    const char* remain = fix_hdr_parser.buf_remain();
    success = sec_list_parser.Parse(std::string_view(remain, std::end(message) - remain));
    if (success) {
      LOG(INFO) << "sec_list_parser failed to parse full message";
    }
    if (__glibc_likely(
            sec_list_parser.IsPresented(tag::common::SecurityRequestResult()) &&
            sec_list_parser.GetVal(tag::common::SecurityRequestResult()) == "0")) {
      remain = sec_list_parser.buf_remain();
      for (auto i = 0; i < sec_list_parser.GetVal(tag::common::NoRelatedSym()); ++i) {
        rpt_grp_parser.Reset();
        if (rpt_grp_parser.Parse(std::string_view(remain, std::end(message) - remain), false)) {
          auto& symbol = rpt_grp_parser.GetVal(tag::common::Symbol());
          auto& contract_multiplier = rpt_grp_parser.GetVal(tag::common::ContractMultiplier());
          if (symbol.empty() == false && contract_multiplier >= 0) {
            this->SetContractValue(std::string(symbol), contract_multiplier);
          }
        }
        if ((remain = rpt_grp_parser.buf_remain()) == nullptr) break;
      }
    } else {
      LOG(WARNING) << "Invalid SecurityRequestResult"
                   << sec_list_parser.GetVal(tag::common::SecurityRequestResult());
    }
  }
  return kNeedFurtherParsing;
}

void FixTradeClient2::EstablishSession() {
  stream_client_->WaitConnected();  // wait for auto reconnect
  LOG(INFO) << "Connected. Send logon.";
  char* msg_buf = nullptr;
  stream_client_->ResetSeqNo(1);
  auto buf_size = stream_client_->GetSendMsgBuf(msg_buf);
  auto* logon = msg_builder_->GenerateLogon(
      msg_buf,
      buf_size,
      ctx_.sender_comp_id,
      ctx_.secret_key,
      ctx_.trade_client_ctx.cancel_orders_on_disconnect());
  if (logon == nullptr) {
    LOG(ERROR) << "failed to generate logon";
    return;
  }
  stream_client_->Send(
      StreamClient::SendObj::Raw{logon->GetBufStart(), logon->GetBufSize()},
      *forwarder_grp_);
}

}  // namespace coin2::exchange::deribit::api_order::fix
